package com.unknownman.block.transaction.subscribe;

import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.List;
import java.util.Vector;
import java.util.concurrent.Callable;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.cloud.core.redis.RedisUtil;
import org.cloud.utils.SpringContextUtil;
import org.cloud.utils.process.ProcessUtil;
import org.web3j.protocol.Web3j;
import org.web3j.protocol.core.DefaultBlockParameter;
import org.web3j.protocol.core.DefaultBlockParameterName;
import org.web3j.protocol.core.methods.response.EthBlock;
import org.web3j.protocol.core.methods.response.EthBlock.Block;
import org.web3j.protocol.core.methods.response.EthBlock.TransactionObject;
import org.web3j.protocol.core.methods.response.EthBlock.TransactionResult;
import org.web3j.protocol.core.methods.response.TransactionReceipt;
import org.web3j.protocol.http.HttpService;
import reactor.core.publisher.Mono;

/**
 * 提供基础的交易订阅监听服务
 */
@Slf4j
public abstract class BaseTranFilter {

    Web3j web3j;
    protected RedisUtil redisUtil;
    protected BigInteger lastBlock;

    @Getter
    private Disposable subscribe;

    String rpcAddress;

    public <T extends Web3j> BaseTranFilter(Class<T> web3jClass) {
        this.web3j = SpringContextUtil.getBean(web3jClass);
        RedisUtil redisUtil = SpringContextUtil.getBean(RedisUtil.class);
        if (redisUtil == null) {
            throw new IllegalStateException("redisUtil 不能为空");
        }
        this.redisUtil = redisUtil;
    }

    public BaseTranFilter() {
        RedisUtil redisUtil = SpringContextUtil.getBean(RedisUtil.class);
        if (redisUtil == null) {
            throw new IllegalStateException("redisUtil 不能为空");
        }
        this.redisUtil = redisUtil;
    }

    public BaseTranFilter(String rpcAddress) {
        this.web3j = Web3j.build(new HttpService(rpcAddress));
        this.rpcAddress = rpcAddress;
    }

    final int blockPerCount = 100; //每次订阅处理的区块

    /**
     * 添加监听
     */
    public boolean addFilter(BigInteger startBlock) {

        lastBlock = startBlock;
        log.info("addFilter: 开始订阅交易, [startBlock: {}]", startBlock);
        List<Callable<Boolean>> callableList = new Vector<>();
        subscribe = web3j.replayPastBlocksFlowable(DefaultBlockParameter.valueOf(startBlock), DefaultBlockParameterName.LATEST, true)
            .doOnError(e -> {
                log.error("订阅错误{}", e.getMessage());
                stopSubscribe();
            }).observeOn(Schedulers.io()).doOnSubscribe(subscription -> {
                log.info("订阅中，订阅阙状态：{}", Mono.just(subscription).subscribe().isDisposed());
            }).subscribe(ethBlock -> {

                if (callableList.size() == 0) {
                    log.info("开始订阅【{}】区块", ethBlock.getBlock().getNumber().longValue());
                }
                synchronized (callableList) {
                    callableList.add(() -> {
                        try {
                            this.executeBlock(ethBlock);
                        } catch (Exception e) {
                            return false;
                        }
                        return true;
                    });
                }

                if (callableList.size() >= 100) {
                    log.info("已经订阅{}区块，结束区块【{}】开始处理", blockPerCount, ethBlock.getBlock().getNumber().longValue());
                    final long startTime = System.currentTimeMillis();
                    executeCallables(callableList);
                    log.info("{}个区块处理时间[{}]ms", blockPerCount, System.currentTimeMillis() - startTime);
                    this.updateLastBlock(ethBlock.getBlock().getNumber().add(BigInteger.ONE), true);
                    callableList.clear();
                }

            }, err -> {
                callableList.clear();
                stopSubscribe();
            });
        return true;
    }

//    public void runBlockCallables(List<Callable<Boolean>> callableList) {
//        List<Boolean> executeResult = ProcessUtil.single().runCablles(callableList, 20, 60 * 60 * 100);
//        List<Callable<Boolean>> callablesFail = new ArrayList<>();
//        for (int i = 0; i < executeResult.size(); i++) {
//            Boolean result = executeResult.get(i);
//            if (result == null || result) {
//                callablesFail.add(callableList.get(i));
//            }
//        }
//        if (callablesFail.size() > 0) {
//            runBlockCallables(callablesFail);
//        }
//    }

    public void executeBlock(EthBlock block) {
        // 获取到所需的块信息
        Block ethBlock = block.getBlock();
        // transaction信息获取
        if (ethBlock.getTransactions() == null || ethBlock.getTransactions().isEmpty()) {
            return;
        }
        List<Callable<Boolean>> callables = new ArrayList<>();
        List<TransactionResult> transactions = ethBlock.getTransactions();
        for (TransactionResult<TransactionObject> transactionResult : transactions) {
            TransactionObject transaction = transactionResult.get();
            // log 数据的获取 记得过滤一下已经removed 的log数据
            callables.add(() -> {
                try {
                    this.subscribeCallBack(getTransactionReceipt(transaction.getHash()));
                } catch (Exception e) {
                    return false;
                }
                return true;
            });
        }
        executeCallables(callables);
//        log.info("[{}]-处理成功！", block.getBlock().getNumber());
    }

    private void executeCallables(List<Callable<Boolean>> callables) {
        List<Boolean> executeResult = ProcessUtil.single().runCablles(callables, blockPerCount, 600 * 1000);
        List<Callable<Boolean>> callablesFail = new ArrayList<>();
        for (int i = 0; i < executeResult.size(); i++) {
            Boolean result = executeResult.get(i);
            if (result == null || (!result)) {
                callablesFail.add(callables.get(i));
            }
        }
        if (callablesFail.size() > 0) {
            executeCallables(callablesFail);
        }
    }

    private TransactionReceipt getTransactionReceipt(String hash) {
        try {
            return web3j.ethGetTransactionReceipt(hash).send().getResult();
        } catch (Exception e) {
            return getTransactionReceipt(hash);
        }
    }

    /**
     * 监听到事件后的回调
     *
     * @param transactionReceipt
     */
    public abstract void subscribeCallBack(TransactionReceipt transactionReceipt) throws Exception;

    /**
     * 更新处理到的区块高度
     *
     * @param block
     */
    public abstract void updateLastBlock(BigInteger block, boolean success);

    /**
     * 订阅是否关闭
     *
     * @return
     */
    public boolean isDisposed() {
        if (subscribe == null) {
            return true;
        }
        return subscribe.isDisposed();
    }

    /**
     * 关闭订阅
     */
    public void stopSubscribe() {
        if (!isDisposed()) {
            subscribe.dispose();
        }
    }

    public <T> T getEventFilter() {
        return (T) this;
    }

}
